package com.moli.iov.down.command.mq.kafka;

import com.moli.iov.down.command.mq.kafka.handle.KafkaMessageHandler;
import com.moli.iov.protocol.codec.DataWrapper;
import com.moli.iov.util.ByteArrayUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

/**
 * <p>
 *  kafka 消费者  上行响应报文 终端响应业处理
 * </p>
 *
 * @author: zql
 * @Date: 2018/11/30 14:40
 */
public class  ConsumerMessageListener implements AcknowledgingMessageListener<String,Object> {

    protected final Logger log = LoggerFactory.getLogger(ConsumerMessageListener.class);

    @Autowired
    private TaskExecutor taskExecutor;

    @Autowired
    private KafkaMessageHandler kafkaMessageHandler;


    @Override
    public void onMessage(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment) {
        try {
            taskExecutor.execute(()->{
                String key = record.key();
                try {
                    DataWrapper dataWrapper = (DataWrapper)record.value();

                    String topic = record.topic();
                    String serialNo = dataWrapper == null ? "" : dataWrapper.getSerialNo();
                    String data = dataWrapper == null || dataWrapper.getAllData() == null ? "" :
                        ByteArrayUtil.toHexString(dataWrapper.getAllData());

                    log.info("Kafka Consumer [上行响应报文] 线程池中处理开始 接收到消息  topic: {}, key：{}，serialNo:{}, allData: {} dataWrapper:{}",
                        topic, key, serialNo, data, dataWrapper);

                    kafkaMessageHandler.handle(topic, key, dataWrapper);

                    log.info("Kafka Consumer [上行响应报文]  线程池中处理结束    接收到消息  key：{}， allData: {}", key, data);

                }catch(Exception e){

                    log.error("afka Consumer [上行响应报文]  线程池中处理异常  vin: {}", key, e);
                }
            });
        }catch(Exception e){
            log.error("kafka 消费者  上行响应报文 终端响应业处理 添加线程池异常", e);
        }finally {
            //手动提交
            acknowledgment.acknowledge();
        }
    }
}
